// Package mq 借鉴github.com/vardius/message-bus 实现的消息队列接口
package mq

import (
	"fmt"
	"reflect"
	"slices"
	"sync"

	"github.com/pkg/errors"
)

type Options struct {
	// 是否将数据持久化到磁盘, 默认为 ture
	Persist bool
}

type IMq interface {
	// Publish 发布消息
	Publish(topic string, opts *Options, args ...any) error
	// Subscribe 订阅消息
	Subscribe(topic, channel string, fn any) error
	// Unsubscribe 取消订阅
	Unsubscribe(topic, channel string) error
	// Close 关闭连接
	Close(topic string) error
}

// defaultQueue 默认消息队列
var defaultQueue IMq = New(100000)

// Fn 订阅函数
type Fn struct {
	callback reflect.Value
	message  chan []reflect.Value
	channel  string
}

// Queue 消息队列
type Queue struct {
	queueSize int
	mux       sync.RWMutex

	// 当订阅不存在，先存放到这里，最大为 10w,当超过 10w,需要落盘，清除现在的所有数据
	pubCache       map[string][][]reflect.Value
	deletePubCache chan struct{}

	topicMap map[string]map[string]*Fn
}

// New 创建消息队列
func New(queueSize int) *Queue {
	if queueSize <= 0 {
		queueSize = 1
	}

	queue := &Queue{
		queueSize:      queueSize,
		pubCache:       make(map[string][][]reflect.Value),
		topicMap:       make(map[string]map[string]*Fn),
		deletePubCache: make(chan struct{}),
	}

	go func() {
		select {
		case <-queue.deletePubCache:
			// 清除缓存
			queue.mux.Lock()
			queue.pubCache = nil
			queue.mux.Unlock()
		}
	}()

	return queue
}

// Publish 发布消息
// 不能采用异步，需要保证消息的顺序
func (q *Queue) Publish(topic string, opts *Options, args ...any) error {
	if opts == nil {
		opts = &Options{Persist: true}
	}

	// 转换 args
	refArgs := make([]reflect.Value, 0, len(args))
	for _, arg := range args {
		refArgs = append(refArgs, reflect.ValueOf(arg))
	}

	q.mux.Lock()
	defer q.mux.Unlock()

	if channels, ok := q.topicMap[topic]; ok {
		for _, channel := range channels {
			channel.message <- refArgs
		}
	} else if opts.Persist { // 不存在
		if q.pubCache[topic] == nil {
			q.pubCache[topic] = make([][]reflect.Value, 0)
		}

		q.pubCache[topic] = append(q.pubCache[topic], refArgs)

		// 超过 10w 落盘
		if len(q.pubCache[topic]) > 100000 {
			// TODO
		}
	}

	return nil
}

// Subscribe 订阅消息,
// topic 主题
// channel 唯一标识一个订阅者
// fn 回调函数
func (q *Queue) Subscribe(topic, channel string, fn any) error {
	q.mux.Lock()
	defer q.mux.Unlock()

	if _, ok := q.topicMap[topic]; !ok {
		q.topicMap[topic] = make(map[string]*Fn)
	}

	if channel == "" {
		channel = "default"
	}

	// 判断channel是否已经存在
	if _, ok := q.topicMap[topic][channel]; ok {
		return errors.New("topic channel 已经存在")
	}

	// 判断fn是否为函数
	if reflect.TypeOf(fn).Kind() != reflect.Func {
		return errors.New("fn不是函数类型")
	}

	f := &Fn{
		callback: reflect.ValueOf(fn),
		message:  make(chan []reflect.Value, q.queueSize),
		channel:  channel,
	}

	go func() {
		defer func() {
			if err := recover(); err != nil {
				fmt.Println(topic, channel, err)
			}
		}()

		for args := range f.message {
			f.callback.Call(args)
		}
	}()

	q.topicMap[topic][channel] = f

	// 清空缓存
	if len(q.topicMap[topic]) > 0 {
		for _, args := range q.pubCache[topic] {
			f.callback.Call(args)
		}
		// 删除缓存
		delete(q.pubCache, topic)
	}

	return nil
}

// Unsubscribe 取消订阅
func (q *Queue) Unsubscribe(topic, channel string) error {
	q.mux.Lock()
	defer q.mux.Unlock()

	// 删除订阅
	channelMap, ok := q.topicMap[topic]
	if !ok {
		return errors.New("topic不存在")
	}

	if _, ok := channelMap[channel]; ok {
		delete(q.topicMap[topic], channel)
	}

	// 删除整个主题
	if len(q.topicMap[topic]) == 0 {
		delete(q.topicMap, topic)
	}

	return nil
}

// Close 关闭连接
func (q *Queue) Close(topic string) error {
	q.mux.Lock()
	defer q.mux.Unlock()

	// 删除主题
	delete(q.topicMap, topic)
	return nil
}

// CheckTopic 检查主题是否合法
func (q *Queue) checkTopic(topic string) bool {
	if slices.Index([]string{"", "realtime"}, topic) != -1 {
		return false
	}

	return true
}

// Default 返回默认消息队列
func Default() IMq {
	return defaultQueue
}

// Publish 发布消息
func Publish(topic string, opts *Options, args ...any) error {
	return defaultQueue.Publish(topic, opts, args...)
}

// Subscribe 订阅消息
func Subscribe(topic, channel string, fn any) error {
	return defaultQueue.Subscribe(topic, channel, fn)
}

// Unsubscribe 取消订阅
func Unsubscribe(topic, channel string) error {
	return defaultQueue.Unsubscribe(topic, channel)
}

// Close 关闭连接
func Close(topic string) error {
	return defaultQueue.Close(topic)
}
